package com.atguigu.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.utils.DimHandler;
import com.atguigu.utils.DruidDSUtil;
import com.atguigu.utils.ThreadPoolUtil;
import lombok.SneakyThrows;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;


import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;


//异步IO获取phoenix维表信息
public abstract class DimAsyncJoinFunction<T> extends RichAsyncFunction<T, T> implements getKeyAndJoinFunc<T>{

       private DruidDataSource druidDataSource;
       private ThreadPoolExecutor threadPoolExecutor;

       private String tableName;

    public DimAsyncJoinFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //获取连接池
        druidDataSource = DruidDSUtil.getDruidDataSource();
        //获得线程池
        threadPoolExecutor = ThreadPoolUtil.getThreadPoolExecutor();




    }

//    public abstract String getKey(T input);//getKey()写成抽象方法，让调用者实现
//    public abstract void join(T input, JSONObject dimInfo);//join(dimInfo) 让调用者自己将查询的一行数据的字段取出封装到javabean里

    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) {
        //input：一条条来的数据，可以是javabean
        threadPoolExecutor.execute(new Runnable() {//创建一个线程或利用一个已经存在的线程来执行任务
            @SneakyThrows
            @Override
            public void run() {
                DruidPooledConnection phoenixconn = druidDataSource.getConnection();

                //todo 根据维度外键查询维表数据
                String key = getKey(input);//getKey()写成抽象方法，让调用者实现
                JSONObject dimInfo = DimHandler.getDimInfo(phoenixconn, tableName, key);

                //todo 将查询的一行数据的字段取出封装到javabean里（需要调用者自己实现）
                if (dimInfo != null) {//有可能根本就查不到，因为phoenix根本没有这个key
                    join(input,dimInfo);//抽象方法，调用者自己实现
                }

                //todo 将补充了维度信息的例如javabean也就是input写回到流里，传递给下游
                resultFuture.complete(Collections.singleton(input));

                //todo 回收phoenix连接
                phoenixconn.close();

            }
        });


    }

    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) {
        System.out.println("超时的数据为："+input);

    }
}
